package cn.bby.query.server.infrastructure.db.repository;

import cn.bby.query.server.domain.service.dto.DataUpdateDetail;
import cn.bby.query.server.domain.service.manager.DelayEventBufManager;
import cn.bby.query.server.infrastructure.db.entity.DelayEventBuf;
import cn.bby.query.server.infrastructure.db.mapper.DelayEventBufMapper;
import com.alibaba.fastjson2.JSONObject;
import jakarta.annotation.Resource;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author ben
 * @since 2025/6/8 23:59
 */
@Service
public class DelayEventBufManagerImpl implements DelayEventBufManager {
    private final RLock lock;
    @Resource
    private DelayEventBufMapper delayEventBufMapper;

    public DelayEventBufManagerImpl(RedissonClient redissonClient) {
        lock = redissonClient.getLock("delay_event_buf_lock");
    }

    @Override
    public void createEvents(Long interfaceId, List<DataUpdateDetail> details) {
        List<DelayEventBuf> list = details.stream()
                .map(event -> new DelayEventBuf().setInterfaceId(interfaceId)
                        .setOpType(event.getOpType())
                        .setData(event.getData().toJSONString())
                        .setTableId(event.getTableId())
                )
                .toList();
        delayEventBufMapper.insert(list);
    }

    @Transactional
    @Override
    public Map<Long, List<DataUpdateDetail>> popReadyEvents(int limit) {
        if (lock.tryLock()) {
            try {
                List<DelayEventBuf> events = delayEventBufMapper.getReadyEvents(limit);
                if (events.isEmpty()) {
                    return Collections.emptyMap();
                }
                delayEventBufMapper.deleteByIds(events.stream().map(DelayEventBuf::getId).toList());

                return events.stream()
                        .collect(
                                Collectors.groupingBy(
                                        DelayEventBuf::getInterfaceId,
                                        Collectors.mapping(
                                                buf -> new DataUpdateDetail()
                                                        .setTableId(buf.getTableId())
                                                        .setOpType(buf.getOpType())
                                                        .setData(JSONObject.parseObject(buf.getData())), Collectors.toList()
                                        )
                                )
                        );
            } finally {
                lock.unlock();
            }
        }
        return Collections.emptyMap();
    }
}
